using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace CoreLibrary.Concurrent {

    public class ConcurrentBlockingQueue<T> : BlockingCollection<T> where T : class {

        public delegate void CompleteCallback();
        public delegate void ErrorDelegate(T task, Func<object> error);
        public delegate void TaskDelegate (T task, CompleteCallback completeCallback);

        public enum Priority {
            High   = 1,
            Normal = 0
        }

        public  int MaxThreads { get; set; }
        public  TaskDelegate  ExecuteTask  { get; set; }
        public  ErrorDelegate ErrorHandler { get; set; }
        private readonly ManualResetEvent _starter = new ManualResetEvent(false);
        private readonly Semaphore _semaphore;
        private int _activeCount;

        public ConcurrentBlockingQueue(int maxThreads) 
             : base(new ConcurrentQueue<T>(), maxThreads) {
               MaxThreads = maxThreads;
               _semaphore = new Semaphore(maxThreads,maxThreads);
               ThreadPool.QueueUserWorkItem(Dispatch, null);
        }

        public void Start() {
            _starter.Set();
        }

        public void Stop() {
            _starter.Reset();
        }

        public List<T> Get(int offset, int max) {
            return this.Skip(offset).Take(max).ToList();
        }

        public bool TryEnqueue(T item) {
            return item != null 
                        && TryAdd(item);
        }  

        public T TryDequeue() {
            return Take();
        }

        protected void Dispatch(object state) {
            while (true) {
                _starter.WaitOne();
                var item = TryDequeue();
                if (item == null) {
                    return;
                }
                Task.Factory.StartNew(() => {
                   _semaphore.WaitOne();
                   Process(item);
                });
            }
        }

        protected void Process(T task) {
            ExecuteTask(task, () => {
                T item2;
                TryTake(out item2);
                _semaphore.Release();
            });
        }

    }
}